package com.bc.commonentity.log;

import com.bc.commoncore.config.ConfigBase;
import com.bc.commoncore.util.JsonUtil;
import com.bc.commoncore.util.ParseHelper;
import com.bc.commoncore.util.SystemUtils;
import com.bc.commonentity.common.ActionDetailLog;
import com.bc.commonentity.common.ActionLog;
import com.bc.commonentity.enums.HttpReturnRnums;
import com.bc.commonentity.enums.LogDetailType;
import com.dianping.cat.Cat;
import com.dianping.cat.message.Transaction;
import com.dianping.cat.message.internal.DefaultTransaction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Collectors;

public class LogMqService {
	private static int taskThreadNum=Runtime.getRuntime().availableProcessors()*2;
	private static ExecutorService executorService=Executors.newFixedThreadPool(taskThreadNum);
	private static volatile LinkedBlockingQueue<QueueItem> queue=new LinkedBlockingQueue<QueueItem>();
	private static long timeout = 1000;
	private static Integer listSize = 100;//list中记录超过100条
	private static Integer reqNum = 100;//每个请求调用次数
	private static int batchSize=5000;//线程池中每个线程每次处理的消息数量
	private static volatile long finishedTotal=0;
	private static volatile List<Long> gcList=new ArrayList<>() ;
	private static volatile List<LogBean> list = new ArrayList<LogBean>();
	private static final Logger logger = LoggerFactory.getLogger(LogMqService.class);
	//elk专用logger
	private static final Logger elkLogger = LoggerFactory.getLogger("elkLogger");
	//利用线程池消费队列里面的消息
	public LogMqService(){
		/**
		if (executorService == null) {
			taskThreadNum = Runtime.getRuntime().availableProcessors()*2; 
			executorService=Executors.newFixedThreadPool(taskThreadNum);
		}**/
	}
	/**
	 * 返回日志队列中日志的数量（cat心跳会用到）
	 * @author zhaohl
	 * @date 20170516
	 * @return
	 */
	public static int getQueueSize(){
		return queue.size();
	}
	/**
	 * 返回日志线程池中活动状态线程数量（cat心跳会用到）
	 * @author zhaohl
	 * @date 20170516
	 * @return
	 */
	public static int getActiveThreadCount(){
		return taskThreadNum-getIdleThreadCount();
	}
	/**
	 * 获取线程池中空闲的线程数量（cat心跳会用到）
	 * @author zhaohl
	 * @date 20170504
	 * @return
	 */
	public static int getIdleThreadCount(){
		ThreadPoolExecutor b=(ThreadPoolExecutor)executorService;
		int res=taskThreadNum-b.getActiveCount();
		if (res<=0) {
			return 0;
		}
		return res;
	}
	/**
	 * 每累计处理10万时gc一次
	 * @author zhaohl
	 * @date 20170504
	 * @param finishedCount
	 */
	private static synchronized void recycleMemory(int finishedCount){
		finishedTotal+=finishedCount;
		long batch=finishedTotal/100000;
		if (batch>0) {
			if (!gcList.contains(batch)) {
				long begin=System.currentTimeMillis();
				System.gc();
				logger.debug("recycleMemory耗时"+(System.currentTimeMillis()-begin));
				gcList.add(batch);
			}
		}
	}
	/**
	 * 线程池中的任务都执行完毕之后，进行gc
	 * @author zhaohl
	 * @date 20170504
	 */
	private static synchronized void finallyRecycleMemory(){
		int idleThreadCount=getIdleThreadCount();
		if (idleThreadCount==taskThreadNum-1) {
			finishedTotal=0;
			gcList.clear();
			list.clear();
			logger.debug("最后进行了gc");
			System.gc();
		}
	}
	/**
	 * 如果线程池中有空闲的线程，则添加任务
	 * @author zhaohl
	 * @date 20170504
	 */
	private static synchronized void refreshWorkers(){
		int idleThreadCount=getIdleThreadCount();
		if (idleThreadCount<=0) {
			return;
		}
		if (idleThreadCount!=taskThreadNum
				&&queue.size()<=batchSize*10) {
			return;
		}
		logger.debug("开始添加任务数"+idleThreadCount);
		for (int i = 0; i < idleThreadCount; i++) {
			executorService.execute(new Runnable() {
				@Override
				public void run() {
					try {
						int tempFinishedTotal=0;
						int freeCount=0;
						int queueItemCount=0;
						List<QueueItem> logCache = new ArrayList<QueueItem>();
						while (true) {
							queue.drainTo(logCache,batchSize);
							if (logCache.size()==0) {
								freeCount++;
								if (freeCount<=3) {
									Thread.currentThread().sleep(5000);
								}else {
									finallyRecycleMemory();
									logger.debug("结束任务"+Thread.currentThread().getId()+";消费日志数量"+tempFinishedTotal);
									break;
								}
							}else {
								queueItemCount=logCache.size();
								tempFinishedTotal+=queueItemCount;
								for (QueueItem queueItem : logCache) {
									consumeLog(queueItem.getLogEngity(), queueItem.getDetailLogs());
								}
								logCache.clear();
								recycleMemory(queueItemCount);	
							}

						}
					} catch (Exception e) {
						logger.error("从队列中批量消费日志时异常:"+e.getMessage());
					}
				}
			});
		}	
	}
	/**
	 * 将日志发送到队列中
	 * @author zhaohl
	 * @date 20170504
	 * @param logEngity
	 * @param detailLogs
	 */
	private static void putLog(ActionLog logEngity,List<ActionDetailLog> detailLogs){
		try {
			long executeTime2 = System.currentTimeMillis();
			queue.put(new QueueItem(logEngity,detailLogs));
			logger.debug("放入日志队列耗时："+(System.currentTimeMillis()-executeTime2));
		} catch (Exception e) {
			logger.error("put请求日志到队列时异常:"+e.getMessage());
		}
	}
	/**
	 * 具体的日志处理逻辑
	 * @author zhaohl
	 * @date 20170504
	 * @param logEngity
	 * @param detailLogs
	 */
	private static void consumeLog(ActionLog logEngity,List<ActionDetailLog> detailLogs){
		writeFile(logEngity,detailLogs);
		putCatLog(logEngity,detailLogs);
		if (detailLogs!=null&&detailLogs.size()>0) {
			if ((logEngity!=null&&logEngity.getExecuteTime().intValue()>timeout)||logEngity==null) {
				addList(detailLogs);
			}
		}
		destoryLog(logEngity,detailLogs);
	}
	/**
	 * 异步将日志写入mq
	 * 
	 * @author zhaohailong
	 * @date 20160627
	 */
	public static void asynSendMessage(ActionLog logEngity,List<ActionDetailLog> detailLogs) {
		synSendMessage(logEngity,detailLogs);
	}
	/**
	 * 执行发送邮件的操作
	 * 
	 * @author houzhangbo
	 * 这里超出了设定list存储的阀值,建议发送邮件，然后清空list
	 */
	private static void doSendMail(){
		List<LogBean> temp=list.stream().filter(k->k.getCallCount()>reqNum).collect(Collectors.toList());
		if(list.size()<listSize&&temp.size()<=0){
			return;
		}
		String body = getHtmlMail();
		SystemUtils.sendAlertHtmlEmail("请求慢的日志", body);
		list.clear();
	}
	/**
	 * 拼接html格式的内容
	 * @author houzhangbo
	 * @return
	 */
	private static String getHtmlMail(){
		String appServerIP = "";
		List<String> ipinfoList = SystemUtils.getLocalIpInfo();
		if(null!=ipinfoList&&ipinfoList.size()>0){
			appServerIP = ipinfoList.get(0);
		}
		StringBuilder titleMin = new StringBuilder("请求慢的日志共有");
		titleMin.append(list.size()).append("条</br>");
		String[] str = getTimes();
		if(null!=str&&str.length==2){
			titleMin.append("开始时间:").append(str[0]).append(",结束时间：").append(str[1]).append("</br>服务器IP:").append(appServerIP);
		}
		//根据type进行排序
		Collections.sort(list);
		StringBuilder sbf = new StringBuilder();
		sbf.append("<html><body><table border=\"1\">");
		sbf.append("<caption style=\"text-align: left;\">")
		.append(titleMin.toString())
		.append("</caption>")
		.append("<thead><tr>")
		.append("<th width=\"80\">行数</th>")
		.append("<th width=\"400\">requestKey</th>")
		.append("<th width=\"100\">类型</th>")
		.append("<th width=\"100\">调用次数</th>")
		.append("<th width=\"100\">平均耗时(毫秒)</th>")
		.append("<th width=\"100\">最大耗时(毫秒)</th>")
		.append("<th width=\"100\">最小耗时(毫秒)</th>")
		.append("</tr></thead>");
		sbf.append("<tbody>");
		int i = 1;
		for(LogBean log:list){
			sbf.append("<tr align=\"left\">");
			sbf.append("<td>").append(i++).append("</td>");
			sbf.append("<td>").append(log.getRequest()).append("</td>");
			sbf.append("<td>").append(LogDetailType.getEnum(log.getType()).desc()).append("</td>");
			sbf.append("<td>").append(log.getCallCount()).append("</td>");
			
			sbf.append("<td>").append(log.getAvgTimes()).append("</td>");
			sbf.append("<td>").append(log.getLongestTimes()).append("</td>");
			sbf.append("<td>").append(log.getShortTimes()).append("</td>");
			sbf.append("</tr>");
		}
		sbf.append("<tbody></table></body></html>");
		return sbf.toString();
	}
	/**
	 * 把log详情添加到list中
	 * 
	 * @author houzhangbo
	 * @param detailLogs
	 */
	private static synchronized void addList(List<ActionDetailLog> detailLogs){
		if(null!=detailLogs&&detailLogs.size()>0){
			//判断当前list是否需要发送邮件
			doSendMail();
			//发送完邮件之后，将本次请求日志加入list
			for(ActionDetailLog log : detailLogs){
				if(log.getExecuteTime().longValue()>timeout){//执行时间大于1s
					//执行添加到list的操作
					doAddList(log);
				}
			}
		}
	}
	/**
	 * 执行添加到list的操作
	 * 
	 * @author houzhangbo
	 * @param log
	 */
	private static void doAddList(ActionDetailLog log){
		if(null!=log&&null!=log.getOptType()){
			if(LogDetailType.service.value()==log.getOptType().intValue()){
				return ;
			}
			
			//获取log中的url
			String requestkey = LogDetailHelper.getRequestKey(log);
			if(org.apache.commons.lang.StringUtils.isBlank(requestkey)){
				return ;
			}
			//判断list中是否存在该项
			LogBean logBean = listHasObj(requestkey,log.getOptType());
			if(null!=logBean){
				//已经存在
				logBean.setCallCount(logBean.getCallCount()+1);
				logBean.getExecTimes().add(log.getExecuteTime());
				logBean.calcu();
				logBean.setExecDate(new Date());
			}else{
				//未存在
				list.add(new LogBean(requestkey,log.getOptType(),log.getExecuteTime()));
			}
			
		}
	}
	/**
	 * 判断list中是否已经存在该项，如果存在就更新调用次数和请求时间
	 * 
	 * @author houzhangbo
	 * @param requestKey
	 * @param type
	 * @return
	 */
	private static LogBean listHasObj(String requestKey,int type){
		if (org.apache.commons.lang.StringUtils.isEmpty(requestKey)) {
			return null;
		}
		if(null!=list&&list.size()>0){
			for(LogBean log:list){
				//已经存在
				if(log.getType().intValue()==type&&requestKey.equals(log.getRequest())){
					return log;
				}
			}
		}
		//未找到，不存在
		return null;
	}
	
	/**
	 * 按照log4j配置的格式,将请求日志写入文本文件，提供给Logstash收集到elk中
	 * @author zhaohl
	 * @date 20170427
	 * @param logEngity
	 * @param detailLogs
	 */
	private static void writeFile(ActionLog logEngity, List<ActionDetailLog> detailLogs){
		if (logEngity!=null) {
			elkLogger.error("access-"+JsonUtil.obj2string(logEngity));
		}
		if (detailLogs!=null&&detailLogs.size()>0) {
			for (ActionDetailLog actionDetailLog : detailLogs) {
				LogDetailType logDetailType=LogDetailType.getEnum(actionDetailLog.getOptType());
				if (logDetailType!=null&&logDetailType!=LogDetailType.service) {
					actionDetailLog.setParentMethodName("");
					if (logDetailType==LogDetailType.db) {
						if (actionDetailLog.getSql()!=null&&!actionDetailLog.getSql().isEmpty()) {
							 actionDetailLog.setSql(actionDetailLog.getSql().replace("\r", " ").replace("\n"," ").replace("\t"," "));
						}
					}
					if (logDetailType==LogDetailType.postget||logDetailType==LogDetailType.redis) {
						actionDetailLog.setRequestkey(LogDetailHelper.getRequestKey(actionDetailLog));
					}
					elkLogger.error(logDetailType.desc()+"-"+JsonUtil.obj2string(actionDetailLog));
				}
			}
		}
	}
	/**
	 * 同步将日志写入mq
	 * 
	 * @author zhaohailong
	 * @date 20160627
	 */
	public static void synSendMessage(ActionLog logEngity,List<ActionDetailLog> detailLogs) {
		if (logEngity==null&&detailLogs==null) {
			return;
		}
		//ConfigBase.getIsOpenESLog()
		//0表示关闭明细日志(logEngity非空,detailLogs为null)
		//1表示开启详细的日志（logEngity和detailLogs都有值）
		//2表示开启详细分布式调用链日志(logEngity和detailLogs都有值)
		//ConfigBase.getIsOpenMongoLog()
		//0表示不写elk
		//1表示写入elk
		if (ConfigBase.getIsOpenMongoLog().equals("0")) {
			destoryLog(logEngity,detailLogs);
			return;
		}

		if (getIdleThreadCount()>0) {
			refreshWorkers();
		}
		putLog(logEngity,detailLogs);
	}
	/**
	 * 处理完日志，清理对象，节省内存
	 * @author zhaohl
	 * @date 20170427
	 * @param logEngity
	 * @param detailLogs
	 */
	private static void destoryLog(ActionLog logEngity,List<ActionDetailLog> detailLogs){
		if (logEngity!=null) {
			logEngity=null;
		}
		if (detailLogs!=null) {
			if (detailLogs.size()>0) {
				detailLogs.clear();
			}
			detailLogs=null;
		}
	}
	/**
	 * 根据list中bean的添加时间获取开始和结束时间
	 * @return
	 */
	private static String[] getTimes(){
		if(null!=list&&list.size()>0){
			Comparator<LogBean> comparator0 =(a, b) -> a.getExecDate().compareTo(b.getExecDate());
			//LogBean minBean=list.stream().min(comparator0).get();
			String[] str = new String[2];
			Optional<LogBean> minObj = list.stream().min(comparator0);
			LogBean minBean = null;
			if (minObj.isPresent()) {
				minBean = minObj.get();
				if (minBean != null) {
					str[0] =ParseHelper.ToDateString(minBean.getExecDate());
				}
			}
			//LogBean maxBean=list.stream().max(comparator0).get();
			Optional<LogBean> maxObj = list.stream().max(comparator0);
			LogBean maxBean = null;
			if (maxObj.isPresent()) {
				maxBean = maxObj.get();
				if (maxBean != null) {
					str[1] =ParseHelper.ToDateString(maxBean.getExecDate());
				}
			}
			return str;
		}
		return null;
	}
	private static String getCauseBy(ActionLog logEngity,ActionDetailLog detail){
		if (logEngity!=null&&logEngity.getHasError()==1) {
			if (logEngity.getException()!=null&&
					!logEngity.getException().isEmpty()&&
					!logEngity.getException().equals(HttpReturnRnums.SystemError.desc())) {
				return ParseHelper.filterCauseBy(logEngity.getException());
			}
			return ParseHelper.filterCauseBy(logEngity.getStackTrace());
		}
		if (detail!=null&&detail.getHasError()==1) {
			return ParseHelper.filterCauseBy(detail.getStackTrace());
		}
		return Transaction.SUCCESS;
	}
	/**
	 * 结束DefaultTransaction
	 * @param t
	 * @param logEngity
	 * @param detail
	 */
	private static void completeTransaction(DefaultTransaction t,ActionLog logEngity,ActionDetailLog detail){
		if (logEngity!=null) {
			t.setStatus(getCauseBy(logEngity,null));
			t.setDurationInMillis(logEngity.getExecuteTime());
		}else if(detail!=null){
			t.setStatus(getCauseBy(null,detail));
			t.setDurationInMillis(detail.getExecuteTime());
		}
		t.setCompleted(true);
		if (Cat.getManager()!=null) {
			Cat.getManager().end(t);
		}
		//DefaultTransaction内部的complete方法会根据纳秒重新计算方法耗时
		//因此这里不能用t.complete()来结束DefaultTransaction
		//t.complete();
	}
	private static void putCatDetailLog(List<ActionDetailLog> detailLogs){
		if (detailLogs!=null&&detailLogs.size()>0) {
			for (ActionDetailLog actionDetailLog : detailLogs) {
				LogDetailType logDetailType=LogDetailType.getEnum(actionDetailLog.getOptType());
				if (logDetailType!=null&&logDetailType!=LogDetailType.service) {
					if (logDetailType==LogDetailType.postget||logDetailType==LogDetailType.redis) {
						actionDetailLog.setRequestkey(LogDetailHelper.getRequestKey(actionDetailLog));
						actionDetailLog.setMethodName(actionDetailLog.getRequestkey());
					}
					Transaction t=Cat.newTransaction(logDetailType.desc(), actionDetailLog.getMethodName());
					if (t instanceof DefaultTransaction) {
						completeTransaction((DefaultTransaction)t,null,actionDetailLog);
					}
				}
			}
		}
	}
	private static void putCatLog(ActionLog logEngity,List<ActionDetailLog> detailLogs){
		return;
//		if(!ConfigBase.getIsOpenESLog().equals("1")){
//			return;
//		}
//		long begin =System.currentTimeMillis();
//		try {
//			if (logEngity!=null) {
//				Transaction t=Cat.newTransaction(logEngity.getAccessType(), LogDetailHelper.getRequestUrl(logEngity.getRequestUrl()));
//				if (t instanceof DefaultTransaction) {
//					putCatDetailLog(detailLogs);
//					completeTransaction((DefaultTransaction)t,logEngity,null);
//				}
//			}else {//定时任务执行时，logEngity为null
//				putCatDetailLog(detailLogs);
//			}
//		} catch (Exception e) {
//			logger.error("put请求日志到cat时异常:"+e.getMessage()+StringUtils.getStackTrace(e));
//		}
//		long ss=System.currentTimeMillis()-begin;
//		if (ss>5) {
//			logger.debug("putCatLog耗时"+ss);
//		}
		
	}
	static class QueueItem{
		private ActionLog logEngity;
		private List<ActionDetailLog> detailLogs;
		public QueueItem(){}
		public QueueItem(ActionLog logEngity,List<ActionDetailLog> detailLogs){
			this.logEngity=logEngity;
			this.detailLogs=detailLogs;
		}
		public ActionLog getLogEngity() {
			return logEngity;
		}
		public void setLogEngity(ActionLog logEngity) {
			this.logEngity = logEngity;
		}
		public List<ActionDetailLog> getDetailLogs() {
			return detailLogs;
		}
		public void setDetailLogs(List<ActionDetailLog> detailLogs) {
			this.detailLogs = detailLogs;
		}
	}
}
